// Copyright 2024 Matrix Origin
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package compile

import (
	"bytes"
	"testing"

	"github.com/matrixorigin/matrixone/pkg/pb/plan"
	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
	"github.com/matrixorigin/matrixone/pkg/sql/colexec/connector"
	"github.com/matrixorigin/matrixone/pkg/sql/colexec/dispatch"
	"github.com/matrixorigin/matrixone/pkg/sql/colexec/insert"
	"github.com/matrixorigin/matrixone/pkg/sql/models"
	"github.com/matrixorigin/matrixone/pkg/util"
	"github.com/matrixorigin/matrixone/pkg/util/trace/impl/motrace/statistic"
	"github.com/matrixorigin/matrixone/pkg/vm"
	"github.com/matrixorigin/matrixone/pkg/vm/process"
	"github.com/stretchr/testify/require"
)

const (
	isFirstTrue = 1 << 0 // 0001 : isFirst = true
	isLastTrue  = 1 << 1 // 0010 : isLast = true

	isFirstFalse = 0 << 0 // 0000 : isFirst = false
	isLastFalse  = 0 << 1 // 0000 : isLast = false
)

func Test_processPhyScope(t *testing.T) {
	/*
		mysql> explain SELECT ROUND(SUM(distance), 2) AS total_distance from real_time_position WHERE time_stamp < 1694733800000;
		+----------------------------------------------------------------------------+
		| AP QUERY PLAN ON MULTICN(4 core)                                           |
		+----------------------------------------------------------------------------+
		| Project [2]                                                                |
		|   ->  Aggregate [1]                                                        |
		|         Aggregate Functions: sum(real_time_position.distance)              |
		|         ->  Table Scan [0]  on cloud_device.real_time_position             |
		|               Filter Cond: (real_time_position.time_stamp < 1694733800000) |
		+----------------------------------------------------------------------------+
		Version: 1.0, S3IOInputCount: 0, S3IOOutputCount: 0
		LOCAL SCOPES:
		Scope 1 (Magic: Merge, mcpu: 1, Receiver: [4])
		  Pipeline: └── Output (idx:-1, isFirst:false, isLast:false)
		                └── Projection (idx:2, isFirst:true, isLast:true)
		                    └── Projection (idx:1, isFirst:false, isLast:true)
		                        └── MergeGroup (idx:1, isFirst:false, isLast:false)
		                            └── Merge (idx:1, isFirst:false, isLast:false)
		  PreScopes: {
		    Scope 2 (Magic: Normal, mcpu: 4, Receiver: [0, 1, 2, 3])
		      Pipeline: └── Connector (idx:1, isFirst:false, isLast:false) to MergeReceiver 4
		                    └── MergeGroup (idx:1, isFirst:false, isLast:false)
		                        └── Merge (idx:1, isFirst:false, isLast:false)
		      PreScopes: {
		        Scope 3 (Magic: Normal, mcpu: 1, Receiver: [])
		          DataSource: cloud_device.real_time_position[time_stamp distance]
		          Pipeline: └── Connector (idx:0, isFirst:false, isLast:false) to MergeReceiver 0
		                        └── Group (idx:1, isFirst:true, isLast:false)
		                            └── Projection (idx:0, isFirst:false, isLast:true)
		                                └── Filter (idx:0, isFirst:false, isLast:false)
		                                    └── TableScan (idx:0, isFirst:true, isLast:false)
		        Scope 4 (Magic: Normal, mcpu: 1, Receiver: [])
		          DataSource: cloud_device.real_time_position[time_stamp distance]
		          Pipeline: └── Connector (idx:0, isFirst:false, isLast:false) to MergeReceiver 1
		                        └── Group (idx:1, isFirst:true, isLast:false)
		                            └── Projection (idx:0, isFirst:false, isLast:true)
		                                └── Filter (idx:0, isFirst:false, isLast:false)
		                                    └── TableScan (idx:0, isFirst:true, isLast:false)
		      }
		  }
	*/

	operatorStats := &process.OperatorStats{
		OperatorName:     "ExampleOperator",
		CallNum:          10,
		TimeConsumed:     5000,
		WaitTimeConsumed: 2000,
		MemorySize:       1024,
		SpillSize:        1024,
		InputRows:        1000,
		OutputRows:       950,
		InputSize:        2048,
		InputBlocks:      0,
		OutputSize:       1900,
		ScanBytes:        0,
		NetworkIO:        600,
		//TotalScanTime:         1500,
		//TotalInsertTime:       0,
	}
	operatorStats.AddOpMetric(process.OpScanTime, 1500)
	operatorStats.AddOpMetric(process.OpInsertTime, 2500)
	operatorStats.AddOpMetric(process.OpIncrementTime, 3500)

	phyOperator1_0 := models.PhyOperator{
		OpName:  "Merge",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator1_1 := models.PhyOperator{
		OpName:   "MergeGroup",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_0},
	}

	phyOperator1_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_1},
	}

	phyOperator1_3 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_2},
	}

	phyOperator1_4 := models.PhyOperator{
		OpName:   "Output",
		NodeIdx:  -1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_3},
	}

	phyScope1 := models.PhyScope{
		Magic: "Merge",
		Receiver: []models.PhyReceiver{
			{
				Idx:        4,
				RemoteUuid: "",
			},
		},
		RootOperator: &phyOperator1_4,
	}

	//---------------------------------------------------------------------------------
	phyOperator2_0 := models.PhyOperator{
		OpName:  "Merge",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator2_1 := models.PhyOperator{
		OpName:   "MergeGroup",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator2_0},
	}

	phyOperator2_2 := models.PhyOperator{
		OpName:  "Connect",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		DestReceiver: []models.PhyReceiver{
			{
				Idx:        4,
				RemoteUuid: "",
			},
		},
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator2_1},
	}

	phyScope2 := models.PhyScope{
		Magic: "Normal",
		Receiver: []models.PhyReceiver{
			{
				Idx:        0,
				RemoteUuid: "",
			},
			{
				Idx:        1,
				RemoteUuid: "",
			},
		},
		RootOperator: &phyOperator2_2,
	}
	//---------------------------------------------------------------------------------

	phyOperator3_0 := models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 0,
		Status:  isFirstTrue | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator3_1 := models.PhyOperator{
		OpName:   "Filter",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_0},
	}

	phyOperator3_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_1},
	}

	phyOperator3_3 := models.PhyOperator{
		OpName:   "Group",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_2},
	}

	phyOperator3_4 := models.PhyOperator{
		OpName:  "Connect",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		DestReceiver: []models.PhyReceiver{
			{
				Idx:        0,
				RemoteUuid: "",
			},
		},
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_3},
	}

	phyScope3 := models.PhyScope{
		Magic:        "Normal",
		PreScopes:    []models.PhyScope{},
		RootOperator: &phyOperator3_4,
	}
	//----------------------------------------------------------------------------
	phyOperator4_0 := models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 0,
		Status:  isFirstTrue | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator4_1 := models.PhyOperator{
		OpName:   "Filter",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_0},
	}

	phyOperator4_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_1},
	}

	phyOperator4_3 := models.PhyOperator{
		OpName:   "Group",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_2},
	}

	phyOperator4_4 := models.PhyOperator{
		OpName:  "Connect",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		DestReceiver: []models.PhyReceiver{
			{
				Idx:        0,
				RemoteUuid: "",
			},
		},
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_3},
	}

	phyScope4 := models.PhyScope{
		Magic:        "Normal",
		RootOperator: &phyOperator4_4,
	}

	phyScope2.PreScopes = []models.PhyScope{phyScope3, phyScope4}
	phyScope1.PreScopes = []models.PhyScope{phyScope2}

	type args struct {
		scope models.PhyScope
		nodes []*plan.Node
	}
	tests := []struct {
		name string
		args args
	}{
		{
			name: "test01",
			args: args{
				scope: phyScope1,
				nodes: []*plan.Node{
					{
						NodeId:      0,
						NodeType:    plan.Node_TABLE_SCAN,
						AnalyzeInfo: new(plan.AnalyzeInfo),
						Children:    []int32{},
					},
					{
						NodeId:      1,
						NodeType:    plan.Node_AGG,
						AnalyzeInfo: new(plan.AnalyzeInfo),
						Children:    []int32{0},
					},
					{
						NodeId:      2,
						NodeType:    plan.Node_PROJECT,
						AnalyzeInfo: new(plan.AnalyzeInfo),
						Children:    []int32{1},
					},
				},
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			phyPlan := models.NewPhyPlan()
			phyPlan.LocalScope = append(phyPlan.LocalScope, tt.args.scope)

			statsInfo := new(statistic.StatsInfo)
			statsInfo.ParseStage.ParseDuration = 72872
			statsInfo.PlanStage.PlanDuration = 7544049
			statsInfo.PlanStage.BuildPlanStatsDuration = 142500
			statsInfo.CompileStage.CompileDuration = 59396
			statsInfo.CompileStage.CompileTableScanDuration = 260717
			statsInfo.CompileStage.CompileS3Request = statistic.S3Request{
				List:      0,
				Get:       2,
				Put:       1,
				Head:      1,
				Delete:    0,
				DeleteMul: 0,
			}

			statsInfo.PrepareRunStage.CompilePreRunOnceDuration = 49396
			statsInfo.PrepareRunStage.ScopePrepareDuration = 12000
			statsInfo.PrepareRunStage.BuildReaderDuration = 11000

			explainPhyPlan := models.ExplainPhyPlan(phyPlan, statsInfo, models.NormalOption)
			t.Logf("%s", explainPhyPlan)
			processPhyScope(&tt.args.scope, nil, tt.args.nodes, new(statistic.StatsInfo))
		})
	}
}

func Test_explainResourceOverview(t *testing.T) {
	operatorStats := &process.OperatorStats{
		OperatorName:     "ExampleOperator",
		CallNum:          10,
		TimeConsumed:     5000,
		WaitTimeConsumed: 2000,
		MemorySize:       1024,
		SpillSize:        1024,
		InputRows:        1000,
		OutputRows:       950,
		InputSize:        2048,
		InputBlocks:      0,
		OutputSize:       1900,
		ScanBytes:        0,
		NetworkIO:        600,
		//TotalScanTime:         1500,
		//TotalInsertTime:       0,
	}
	operatorStats.AddOpMetric(process.OpScanTime, 1500)
	operatorStats.AddOpMetric(process.OpInsertTime, 2500)
	operatorStats.AddOpMetric(process.OpIncrementTime, 3500)

	phyOperator1_0 := models.PhyOperator{
		OpName:  "Merge",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator1_1 := models.PhyOperator{
		OpName:   "MergeGroup",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_0},
	}

	phyOperator1_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_1},
	}

	phyOperator1_3 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_2},
	}

	phyOperator1_4 := models.PhyOperator{
		OpName:   "Output",
		NodeIdx:  -1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_3},
	}

	phyScope1 := models.PhyScope{
		Magic: "Merge",
		Receiver: []models.PhyReceiver{
			{
				Idx:        4,
				RemoteUuid: "",
			},
		},
		RootOperator: &phyOperator1_4,
	}

	//---------------------------------------------------------------------------------
	phyOperator2_0 := models.PhyOperator{
		OpName:  "Merge",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator2_1 := models.PhyOperator{
		OpName:   "MergeGroup",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator2_0},
	}

	phyOperator2_2 := models.PhyOperator{
		OpName:  "Connect",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		DestReceiver: []models.PhyReceiver{
			{
				Idx:        4,
				RemoteUuid: "",
			},
		},
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator2_1},
	}

	phyScope2 := models.PhyScope{
		Magic: "Normal",
		Receiver: []models.PhyReceiver{
			{
				Idx:        0,
				RemoteUuid: "",
			},
			{
				Idx:        1,
				RemoteUuid: "",
			},
		},
		RootOperator: &phyOperator2_2,
	}
	//---------------------------------------------------------------------------------

	phyOperator3_0 := models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 0,
		Status:  isFirstTrue | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator3_1 := models.PhyOperator{
		OpName:   "Filter",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_0},
	}

	phyOperator3_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_1},
	}

	phyOperator3_3 := models.PhyOperator{
		OpName:   "Group",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_2},
	}

	phyOperator3_4 := models.PhyOperator{
		OpName:  "Connect",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		DestReceiver: []models.PhyReceiver{
			{
				Idx:        0,
				RemoteUuid: "",
			},
		},
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator3_3},
	}

	phyScope3 := models.PhyScope{
		Magic:        "Normal",
		PreScopes:    []models.PhyScope{},
		RootOperator: &phyOperator3_4,
	}
	//----------------------------------------------------------------------------
	phyOperator4_0 := models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 0,
		Status:  isFirstTrue | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator4_1 := models.PhyOperator{
		OpName:   "Filter",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_0},
	}

	phyOperator4_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  0,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_1},
	}

	phyOperator4_3 := models.PhyOperator{
		OpName:   "Group",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_2},
	}

	phyOperator4_4 := models.PhyOperator{
		OpName:  "Connect",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		DestReceiver: []models.PhyReceiver{
			{
				Idx:        0,
				RemoteUuid: "",
			},
		},
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator4_3},
	}

	phyScope4 := models.PhyScope{
		Magic:        "Normal",
		RootOperator: &phyOperator4_4,
	}

	phyScope2.PreScopes = []models.PhyScope{phyScope3, phyScope4}
	phyScope1.PreScopes = []models.PhyScope{phyScope2}

	statsInfo := new(statistic.StatsInfo)
	statsInfo.ParseStage.ParseDuration = 72872
	statsInfo.PlanStage.PlanDuration = 7544049
	statsInfo.PlanStage.BuildPlanStatsDuration = 142500

	statsInfo.CompileStage.CompileDuration = 59396
	statsInfo.CompileStage.CompileTableScanDuration = 260717
	statsInfo.CompileStage.CompileS3Request = statistic.S3Request{
		List:      0,
		Get:       2,
		Put:       1,
		Head:      1,
		Delete:    0,
		DeleteMul: 0,
	}

	statsInfo.PrepareRunStage.CompilePreRunOnceDuration = 49396
	statsInfo.PrepareRunStage.ScopePrepareDuration = 12000
	statsInfo.PrepareRunStage.BuildReaderDuration = 11000

	type args struct {
		queryResult *util.RunResult
		statsInfo   *statistic.StatsInfo
		anal        *AnalyzeModule
		option      *ExplainOption
		buffer      *bytes.Buffer
	}
	tests := []struct {
		name string
		args args
	}{
		{
			name: "test01",
			args: args{
				queryResult: &util.RunResult{
					AffectRows: 1,
				},
				statsInfo: statsInfo,
				anal: &AnalyzeModule{
					phyPlan: &models.PhyPlan{},
				},
				option: &ExplainOption{
					Analyze: true,
				},
				buffer: bytes.NewBuffer(make([]byte, 0, 300)),
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			phyPlan := models.NewPhyPlan()
			phyPlan.LocalScope = append(phyPlan.LocalScope, phyScope1)
			tt.args.anal.phyPlan = phyPlan
			explainResourceOverview(tt.args.queryResult, tt.args.statsInfo, tt.args.anal, tt.args.option, tt.args.buffer)
			t.Logf("%s", tt.args.buffer.String())
		})
	}
}

func Test_UpdatePreparePhyScope(t *testing.T) {
	operatorStats := &process.OperatorStats{
		OperatorName:     "ExampleOperator",
		CallNum:          10,
		TimeConsumed:     5000,
		WaitTimeConsumed: 2000,
		MemorySize:       1024,
		SpillSize:        1024,
		InputRows:        1000,
		OutputRows:       950,
		InputSize:        2048,
		InputBlocks:      0,
		OutputSize:       1900,
		ScanBytes:        0,
		NetworkIO:        600,
		//TotalScanTime:         1500,
		//TotalInsertTime:       0,
	}
	operatorStats.AddOpMetric(process.OpScanTime, 1500)
	operatorStats.AddOpMetric(process.OpInsertTime, 2500)
	operatorStats.AddOpMetric(process.OpIncrementTime, 3500)

	phyOperator1_0 := models.PhyOperator{
		OpName:  "Merge",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		OpStats: operatorStats,
	}

	phyOperator1_1 := models.PhyOperator{
		OpName:   "MergeGroup",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_0},
	}

	phyOperator1_2 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  1,
		Status:   isFirstFalse | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_1},
	}

	phyOperator1_3 := models.PhyOperator{
		OpName:   "Projection",
		NodeIdx:  1,
		Status:   isFirstTrue | isLastTrue,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_2},
	}

	phyOperator1_4 := models.PhyOperator{
		OpName:   "Output",
		NodeIdx:  -1,
		Status:   isFirstFalse | isLastFalse,
		OpStats:  operatorStats,
		Children: []*models.PhyOperator{&phyOperator1_3},
	}

	phyScope1 := models.PhyScope{
		Magic: "Merge",
		Receiver: []models.PhyReceiver{
			{
				Idx:        4,
				RemoteUuid: "",
			},
		},
		RootOperator: &phyOperator1_4,
	}

	testCompile := NewMockCompile(t)
	var reg process.WaitRegister
	testCompile.proc.Reg.MergeReceivers = []*process.WaitRegister{&reg}
	s2 := generateScopeWithRootOperator(
		testCompile.proc,
		[]vm.OpType{vm.TableScan, vm.Projection, vm.Connector})
	res := UpdatePreparePhyScope(s2, phyScope1)
	require.Equal(t, false, res)
}

func TestHandleTailNodeReceiver(t *testing.T) {
	buf := bytes.NewBuffer(make([]byte, 0, 300))
	mp := make(map[*process.WaitRegister]int)
	reg := process.WaitRegister{}
	mp[&reg] = 0
	{
		buf.Reset()
		node := &connector.Connector{
			Reg: &reg,
		}
		hanldeTailNodeReceiver(node, mp, buf)
		require.Equal(t, " to MergeReceiver [0]", buf.String())
	}

	{
		buf.Reset()
		node := &dispatch.Dispatch{
			LocalRegs: []*process.WaitRegister{&reg},
			FuncId:    dispatch.ShuffleToAllFunc,
			RemoteRegs: []colexec.ReceiveInfo{
				{
					NodeAddr: "127.0.0.1:1234",
				},
				{
					NodeAddr: "127.0.0.1:1235",
				},
			},
		}
		hanldeTailNodeReceiver(node, mp, buf)
		require.Equal(t,
			" shuffle to all of MergeReceiver [0] cross-cn receiver info: "+
				"[addr: 127.0.0.1:1234, uuid: 00000000-0000-0000-0000-000000000000], "+
				"[addr: 127.0.0.1:1235, uuid: 00000000-0000-0000-0000-000000000000]",
			buf.String())
	}

	{
		buf.Reset()
		node := &dispatch.Dispatch{
			LocalRegs: []*process.WaitRegister{&reg},
			FuncId:    dispatch.SendToAllFunc,
		}
		hanldeTailNodeReceiver(node, mp, buf)
		require.Equal(t,
			" to all of MergeReceiver [0]",
			buf.String())
	}

	{
		buf.Reset()
		node := &dispatch.Dispatch{
			LocalRegs: []*process.WaitRegister{&reg},
			FuncId:    dispatch.SendToAnyLocalFunc,
		}
		hanldeTailNodeReceiver(node, mp, buf)
		require.Equal(t,
			" to any of MergeReceiver [0]",
			buf.String())
	}

	{
		buf.Reset()
		node := &dispatch.Dispatch{
			LocalRegs: []*process.WaitRegister{&reg},
			FuncId:    dispatch.SendToAnyFunc,
		}
		hanldeTailNodeReceiver(node, mp, buf)
		require.Equal(t,
			" unknow type dispatch [0]",
			buf.String())
	}
}

func TestShowPipelineTree(t *testing.T) {
	buf := bytes.NewBuffer(make([]byte, 0, 300))
	mp := make(map[*process.WaitRegister]int)
	reg := process.WaitRegister{}
	mp[&reg] = 0

	op := dupOperator(
		insert.NewPartitionInsert(
			&insert.Insert{},
			1,
		),
		0,
		0,
	)

	op.GetOperatorBase().OpAnalyzer = process.NewAnalyzer(
		op.GetOperatorBase().Idx,
		op.GetOperatorBase().IsFirst,
		op.GetOperatorBase().IsLast,
		"partition_insert",
	)

	{
		buf.Reset()
		ShowPipelineTree(op, "", true, true, mp, AnalyzeLevel, buf)
		require.Contains(t, buf.String(), "Pipeline: └── unknown")
		require.Contains(t, buf.String(), "TimeCost:0ns")
	}

	{
		buf.Reset()
		ShowPipelineTree(op, "", false, true, mp, VerboseLevel, buf)
		require.Contains(t, buf.String(), "└── unknown")
	}

	{
		buf.Reset()
		ShowPipelineTree(op, "", false, false, mp, VerboseLevel, buf)
		require.Contains(t, buf.String(), "├── unknown")
	}

}

func TestApplyOpStatsToNode_ReadSize(t *testing.T) {
	// Create a query with nodes
	qry := &plan.Query{
		Nodes: []*plan.Node{
			{
				NodeId: 0,
			},
			{
				NodeId: 1,
			},
		},
	}

	// Create operator stats with ReadSize, S3ReadSize, DiskReadSize
	opStats1 := &process.OperatorStats{
		ReadSize:     1024 * 1024, // 1MB
		S3ReadSize:   512 * 1024,  // 0.5MB
		DiskReadSize: 256 * 1024,  // 0.25MB
		ScanBytes:    2048 * 1024, // 2MB
		NetworkIO:    100 * 1024,  // 100KB
		InputBlocks:  10,
	}

	opStats2 := &process.OperatorStats{
		ReadSize:     2048 * 1024, // 2MB
		S3ReadSize:   1536 * 1024, // 1.5MB
		DiskReadSize: 512 * 1024,  // 0.5MB
		ScanBytes:    4096 * 1024, // 4MB
		NetworkIO:    200 * 1024,  // 200KB
		InputBlocks:  20,
	}

	// Create phy operators
	phyOp1 := &models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 0,
		Status:  isFirstFalse | isLastFalse,
		OpStats: opStats1,
	}

	phyOp2 := &models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 0,
		Status:  isFirstFalse | isLastFalse,
		OpStats: opStats2,
	}

	// Initialize AnalyzeInfo for nodes
	qry.Nodes[0].AnalyzeInfo = &plan.AnalyzeInfo{}

	// Create ParallelScopeInfo
	parallelInfo := NewParallelScopeInfo()

	// Apply stats from first operator
	applyOpStatsToNode(phyOp1, qry, qry.Nodes, parallelInfo)

	// Verify first aggregation
	require.Equal(t, int64(1024*1024), qry.Nodes[0].AnalyzeInfo.ReadSize, "ReadSize should be aggregated")
	require.Equal(t, int64(512*1024), qry.Nodes[0].AnalyzeInfo.S3ReadSize, "S3ReadSize should be aggregated")
	require.Equal(t, int64(256*1024), qry.Nodes[0].AnalyzeInfo.DiskReadSize, "DiskReadSize should be aggregated")
	require.Equal(t, int64(2048*1024), qry.Nodes[0].AnalyzeInfo.ScanBytes, "ScanBytes should be aggregated")
	require.Equal(t, int64(100*1024), qry.Nodes[0].AnalyzeInfo.NetworkIO, "NetworkIO should be aggregated")
	require.Equal(t, int64(10), qry.Nodes[0].AnalyzeInfo.InputBlocks, "InputBlocks should be aggregated")

	// Apply stats from second operator (accumulation)
	applyOpStatsToNode(phyOp2, qry, qry.Nodes, parallelInfo)

	// Verify accumulation
	require.Equal(t, int64(3072*1024), qry.Nodes[0].AnalyzeInfo.ReadSize, "ReadSize should accumulate")
	require.Equal(t, int64(2048*1024), qry.Nodes[0].AnalyzeInfo.S3ReadSize, "S3ReadSize should accumulate")
	require.Equal(t, int64(768*1024), qry.Nodes[0].AnalyzeInfo.DiskReadSize, "DiskReadSize should accumulate")
	require.Equal(t, int64(6144*1024), qry.Nodes[0].AnalyzeInfo.ScanBytes, "ScanBytes should accumulate")
	require.Equal(t, int64(300*1024), qry.Nodes[0].AnalyzeInfo.NetworkIO, "NetworkIO should accumulate")
	require.Equal(t, int64(30), qry.Nodes[0].AnalyzeInfo.InputBlocks, "InputBlocks should accumulate")

	// Test with a different node
	phyOp3 := &models.PhyOperator{
		OpName:  "TableScan",
		NodeIdx: 1,
		Status:  isFirstFalse | isLastFalse,
		OpStats: opStats1,
	}

	qry.Nodes[1].AnalyzeInfo = &plan.AnalyzeInfo{}
	applyOpStatsToNode(phyOp3, qry, qry.Nodes, parallelInfo)

	// Verify node 1 has its own stats
	require.Equal(t, int64(1024*1024), qry.Nodes[1].AnalyzeInfo.ReadSize, "Node 1 ReadSize should be separate")
	require.Equal(t, int64(512*1024), qry.Nodes[1].AnalyzeInfo.S3ReadSize, "Node 1 S3ReadSize should be separate")
	require.Equal(t, int64(256*1024), qry.Nodes[1].AnalyzeInfo.DiskReadSize, "Node 1 DiskReadSize should be separate")

	// Verify node 0 stats are unchanged
	require.Equal(t, int64(3072*1024), qry.Nodes[0].AnalyzeInfo.ReadSize, "Node 0 ReadSize should remain unchanged")
}
